package com.ironman.sailboat.mobile.runner;

import cn.hutool.core.date.DateUtil;
import com.ironman.sailboat.mobile.AbstractMqttClient;
import com.ironman.sailboat.mobile.threadpool.ThreadPool;
import com.mybatisflex.core.row.Row;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.noear.snack.ONode;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Function;

@Slf4j
public abstract class AbstractMockRunner extends AbstractMqttClient {

    public static final CountDownLatch COUNT_DOWN_LATCH = new CountDownLatch(4);

    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();

    private final ReentrantReadWriteLock.WriteLock writeLock = rwLock.writeLock();

    @Getter
    private static Date startTime;

    @Getter
    private static Date endTime;

    public void setStartAndEndTime(List<Row> rows) {
        if (!rows.isEmpty()) {
            writeLock.lock();
            try {
                setStartTime(rows.get(0).getDate("ts"));
                setEndTime(rows.get(rows.size() - 1).getDate("ts"));
            } finally {
                writeLock.unlock();
            }
        }
    }

    private void setStartTime(Date ts) {
        if (startTime == null) {
            startTime = ts;
        } else {
            int compare = DateUtil.compare(startTime, ts);
            if (compare < 0) {
                startTime = ts;
            }
        }
    }

    private void setEndTime(Date ts) {
        if (endTime == null) {
            endTime = ts;
        } else {
            int compare = DateUtil.compare(endTime, ts);
            if (compare > 0) {
                endTime = ts;
            }
        }
    }

    public void traceBackData(Function<Date, ONode> function) {
        AtomicInteger counter = new AtomicInteger(0);
        ThreadPool.SCHEDULED_EXECUTOR.scheduleWithFixedDelay(() -> {
            try {
                int andIncrement = counter.getAndIncrement();
                Date ts = DateUtil.offsetSecond(startTime, andIncrement);
                if (DateUtil.compare(ts, endTime) >= 0) {
                    log.warn("数据已用完");
                    return;
                }
                ONode data = function.apply(ts);
                if (data == null) {
                    return;
                }
                publish("mobile/up/original/{}/{}/{}", data);
            } catch (Exception e) {
                log.error("{}", e.getMessage(), e);
            }
        }, 0, 1, TimeUnit.SECONDS);
    }
}
